/**
 * 项目名称：quickstart-storm 文件名：WordNormalizer.java 版本信息： 日期：2017年9月3日 Copyright yangzl Corporation
 * 2017 版权所有 *
 */
package org.quickstart.storm.example2;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

/**
 * WordNormalizer
 *
 * @author：youngzil@163.com
 * @2017年9月3日 下午12:09:11
 * @since 1.0
 */
public class WordNormalizer implements IRichBolt {

  private OutputCollector collector;

  @Override
  public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
    this.collector = collector;
  }

  /**
   * 这是bolt中最重要的方法，每当接收到一个tuple时，此方法便被调用 这个方法的作用就是把文本文件中的每一行切分成一个个单词，并把这些单词发射出去（给下一个bolt处理）
   **/
  @Override
  public void execute(Tuple input) {
    String sentence = input.getString(0);
    String[] words = sentence.split(" ");
    for (String word : words) {
      word = word.trim();
      if (!word.isEmpty()) {
        word = word.toLowerCase();
        // Emit the word
        List a = new ArrayList();
        a.add(input);
        collector.emit(a, new Values(word));
      }
    }
    // 确认成功处理一个tuple
    collector.ack(input);
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declare(new Fields("word"));

  }

  @Override
  public void cleanup() {
    // TODO Auto-generated method stub

  }

  @Override
  public Map<String, Object> getComponentConfiguration() {
    // TODO Auto-generated method stub
    return null;
  }
}
